package com.zzk.www.service.impl;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.zzk.www.domain.Message;
import com.zzk.www.domain.User;
import com.zzk.www.domain.enums.MessageStatus;
import com.zzk.www.nettyServer.server.ChatServer;
import com.zzk.www.nettyServer.tool.ChannelUserHolder;
import com.zzk.www.service.MessageHelperService;
import com.zzk.www.service.MessageDaoService;
import com.zzk.www.tool.cache.QueueService;
import com.zzk.www.tool.cache.CacheService;
import com.zzk.www.tool.common.*;
import com.zzk.www.tool.constants.CacheConstant;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.apache.commons.lang3.StringUtils;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;

@Service
public class MessageHelperServiceImpl implements MessageHelperService {

    private static final ObjectMapper MAPPER = new ObjectMapper();

    private static final Logger logger = Logger.getLogger(MessageHelperServiceImpl.class);


    @Autowired
    private QueueService queueService;

    @Autowired
    private CacheService cacheService;

    @Autowired
    private MessageDaoService messageDaoService;


    @Value("${msg.batchStoreSize}")
    private Integer batchStoreSize;

    /**
     * 是否可以处理消息
     */
    private boolean canProcessMsg = true;


    @Override
    public void backgroudTaskProcess() {
        ExecutorService ec = Executors.newFixedThreadPool(1);

        // 发送消息存储处理
        ec.execute(new Runnable() {
            @Override
            public void run() {
                doStoreMessage();
            }
        });

        ec.shutdown();

    }


    //验证用户(在握手成功后执行回调接口代码)
    @Override
    public void validate(ChannelHandlerContext ctx, FullHttpRequest request) {

        String uri = request.getUri();
        String token = "";
        int index = uri.indexOf("?token=");

        String address = ctx.channel().remoteAddress().toString();

        //匿名用户
        if (index == -1 || StringUtils.isEmpty(uri.substring(index))) {


            User user = new User();
            user.setAddress(address);
            user.setRegist(false);
            user.setUserId(UUID.randomUUID().toString());

            //登录成功
            ChannelUserHolder.put(ctx.channel(), user);

            sendInfoMessage(ctx.channel(),"登陆成功,你当前为匿名用户");

        } else {

            token = uri.substring(index+"?token=".length());
            String tokenKey = CacheConstant.Auth.token + "_" + token;
            User user = (User) BeanHolder.get(CacheService.class, "redisCacheService").get(tokenKey);

            if (user == null) {
                //匿名用户
                sendInfoMessage(ctx.channel(),"登陆成功,你当前为匿名用户");


                user = new User();
                user.setAddress(address);
                user.setRegist(false);
                user.setUserId(UUID.randomUUID().toString());

                ChannelUserHolder.put(ctx.channel(), user);
            } else {
                //注册用户
                user.setRegist(true);

                //如果已经登陆,关闭上个连接
                Channel connectedChannel = ChannelUserHolder.getChannelByUserId(user.getUserId());
                if (connectedChannel != null) {
                    sendInfoMessage(connectedChannel,"你的账号已在别的地点登陆,请重新登陆");
                    connectedChannel.close();
                    ChannelUserHolder.remove(connectedChannel);
                }
                sendInfoMessage(ctx.channel(),"登陆成功,你的姓名为["+user.getUserName()+"]用户id为["+user.getUserId()+"]");

                //登录成功
                ChannelUserHolder.put(ctx.channel(), user);

                handleOffLineMsg(ctx, user);
            }

        }

    }

    //处理离线消息
    private void handleOffLineMsg(ChannelHandlerContext ctx, User user) {
        String offLineListKey = CacheConstant.Msg.userOffLineMsgList + "_" + user.getUserId();
        String offLineMsg;
        while (ctx.channel().isOpen() && StringUtils.isNotEmpty(offLineMsg = queueService.pop(offLineListKey))) {
            //发送消息
            sendMessage(ctx.channel(),offLineMsg,user.getUserId());
        }

    }


    //接收到消息
    @Override
    public void receiveMessage(ChannelHandlerContext ctx, TextWebSocketFrame msg) {
        Channel channel = ctx.channel();
        Message msgBean;
        try {
            msgBean = MAPPER.readValue(msg.text(), Message.class);
        } catch (IOException e) {
            channel.writeAndFlush(new TextWebSocketFrame("消息格式不正确"));
            throw new RuntimeException("============消息格式不正确============" + msg.text());
        }

        if (!canProcessMsg) {
            channel.writeAndFlush(new TextWebSocketFrame("服务器升级处理,拒绝消息发送"));
            return;
        }


        // 获取发送用户信息
        String sendUserKey = CacheConstant.Auth.loginUser + "_" + msgBean.getSendUserId();
        User sendUser = (User) cacheService.get(sendUserKey);

        msgBean.setSendNickname(sendUser.getNickname());
        msgBean.setSendTime(new Date().getTime());


        // 获取接收用户信息
        List<String> receiveUserIds = msgBean.getReceiveUserIds();
        if (msgBean.getReceiveUserId() != null) {
            receiveUserIds.add(msgBean.getReceiveUserId());
        }

        // 发送多个用户处理
        String listKey;
        for (String receiveUserId : receiveUserIds) {

            Channel channel1 = ChannelUserHolder.getChannelByUserId(receiveUserId);
            String sendMsg = JsonUtils.object2Json(msgBean);

            if (channel1 == null) {
                listKey = CacheConstant.Msg.userOffLineMsgList + "_" + receiveUserId;
                queueService.add(listKey, sendMsg);
                return;
            } else {
                sendMessage(channel,sendMsg,sendUser.getUserId());
            }
        }
    }

    //批量储存消息到数据库
    protected void doStoreMessage() {

        QueueService queueService = BeanHolder.get(QueueService.class);
        System.out.println(queueService);


        String listKey = CacheConstant.Msg.sentMsgList + "_" + SystemUtil.getIp();
        List<Message> dataList = new ArrayList<>(350);

        while (canProcessMsg) {
            try {
                final String message = this.queueService.pop(listKey);
                if (message == null) {
                    try {
                        Thread.sleep(2000);
                    } catch (Exception e) {
                    }
                    continue;
                }
                Message msgBean = JsonUtils.json2Object(message, Message.class);
                msgBean.setStatus(MessageStatus.sent.getCode());
                dataList.add(msgBean);
                if (dataList.size() >= batchStoreSize) {
                    int size = dataList.size();
                    messageDaoService.insertBatch(dataList);
                    dataList.clear();
                }
            } catch (Exception e) {
                logger.error("数据存储任务处理出错", e);
            }
        }

        int size = dataList.size();
        if (size > 0) {
            messageDaoService.insertBatch(dataList);
            logger.info("剩余消息已经储存,size=" + size);
        }

    }


    //用户连接(在验证之后调用)
    @Override
    public void userConnected(ChannelHandlerContext ctx) {
    }

    //用户断开
    @Override
    public void userDisconnected(ChannelHandlerContext ctx) {
        ChannelUserHolder.remove(ctx.channel());
    }

    @Override
    public void destory() {
        try{
            BeanHolder.get(ChatServer.class).destory();
            canProcessMsg = false; // 停止处理消息
            logger.info("服务器升级处理,资源释放");
        }catch(Exception e){
            logger.info("服务器关闭失败"+e);
        }

    }

    void sendInfoMessage(Channel channel,String infoMsg){
        channel.writeAndFlush(new TextWebSocketFrame(infoMsg));

    }

    void sendMessage(Channel channel,String jsonMsg,String senderId){

        channel.writeAndFlush(new TextWebSocketFrame(jsonMsg));

        //发送成功，将消息放到已发送队列
        String sentMsgListKey = CacheConstant.Msg.sentMsgList + "_" + senderId;
        queueService.add(sentMsgListKey, jsonMsg);

    }


}
